package xin.alum.aim.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class RocketMqProducer {

    private final  RocketMQConfig config;
    private final Producer producer;
    private final ClientServiceProvider provider;
    public RocketMqProducer(RocketMQConfig config,ClientServiceProvider provider,Producer producer) {
        this.config=config;
        this.producer=producer;
        this.provider=provider;
    }

    /**
     * 发布消息
     */
    public void sendAsync(String tag, String  message) {
        sendAsync(tag,message.getBytes(StandardCharsets.UTF_8));
    }

    /**
     * 发布消息
     */
    public void sendAsync(String tag, byte[] body) {
            Message message = provider.newMessageBuilder().setTopic(config.topic).setTag(tag).setBody(body).build();
            CompletableFuture<SendReceipt> future = producer.sendAsync(message);
            ExecutorService sendCallbackExecutor = Executors.newCachedThreadPool();
            future.whenCompleteAsync((sendReceipt, throwable) -> {
                if (null != throwable) {
                    log.error("Failed to send message", throwable);
                } else if(log.isDebugEnabled()) {
                    log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
                }
            }, sendCallbackExecutor);
    }

    /**
     * 同步消息
     */
    public void send(String tag, byte[] body) {
        Message message = provider.newMessageBuilder().setTopic(config.topic).setTag(tag).setBody(body).build();
        try {
            producer.send(message);
        } catch (Throwable var13) {
            log.error("Failed to send message", var13);
        }
    }
}
